package com.hmeter.hadoop.hdfs;

import java.io.FileInputStream;

import org.apache.hadoop.conf.Configuration;
import org.apache.jmeter.samplers.AbstractSampler;
import org.apache.jmeter.samplers.Entry;
import org.apache.jmeter.samplers.SampleResult;
import org.apache.jmeter.testbeans.TestBean;

import com.hmeter.hadoop.hdfs.HDFSUtil.FileContent;

/**
 * Sampler for interaction with HDFS
 * 
 * @author dino.keco
 * @author bakir.jusufbegovic
 * @version 1.0
 * @since 1.0
 * 
 */
public class HDFSSampler extends AbstractSampler implements TestBean {

	private static final long serialVersionUID = 8190453552987404875L;

	private String nameNode;
	private String path;

	private Boolean createActive;
	private String content;
	private String upload;
	private Boolean overwrite;
	private String createCodec;

	private Boolean deleteActive;
	private Boolean recursive;

	private Boolean renameActive;
	private String newName;

	private Boolean tailActive;
	private long offset;
	private int size;
	private String tailCodec;
	private Boolean info;

	public HDFSSampler() {
		setName(HDFSSampler.class.getSimpleName());
	}

	@Override
	public SampleResult sample(Entry entry) {
		SampleResult result = new SampleResult();
		result.setSampleLabel(getName());
		result.setDataType(SampleResult.TEXT);

		result.sampleStart();
		result.setContentType("application/xml");
		result.setEncodingAndType("UTF-8");

		StringBuilder builder = new StringBuilder();
		String error = null;
		if ((error = validate()) != null) {
			builder.append("<response>");
			builder.append("<error>").append(error).append("</error>");
			builder.append("</response>");
			result.setResponseData(builder.toString().getBytes());
			result.setSuccessful(false);
		} else {
			try {
				Configuration conf = new Configuration();
				conf.set("fs.defaultFS", nameNode);
				conf.set("fs.default.name", nameNode);

				if (createActive.booleanValue()) {
					if (upload != null && !"".equals(upload)) {
						HDFSUtil.createFile(conf, path, new FileInputStream(upload), overwrite, createCodec);
					} else {
						HDFSUtil.createFile(conf, path, content, overwrite, createCodec);
					}
				} else if (deleteActive.booleanValue()) {
					HDFSUtil.deleteFile(conf, path, recursive);
				} else if (renameActive.booleanValue()) {
					HDFSUtil.renameFile(conf, path, newName);
				} else if (tailActive.booleanValue()) {
					FileContent fc = HDFSUtil.tailFile(conf, path, tailCodec, offset, size, info);
					result.setResponseData(fc.toXMLString().getBytes());
				}

				result.setSuccessful(true);
			} catch (Exception e) {
				builder.append("<response>");
				builder.append("<error>").append(e.getMessage()).append("</error>");
				builder.append("</response>");
				result.setResponseData(builder.toString().getBytes());
				result.setSuccessful(false);
			}
		}

		result.sampleEnd();

		return result;
	}

	private String validate() {
		boolean[] actions = new boolean[4];
		actions[0] = createActive.booleanValue();
		actions[1] = deleteActive.booleanValue();
		actions[2] = renameActive.booleanValue();
		actions[3] = tailActive.booleanValue();

		int count = 0;
		for (boolean a : actions) {
			if (a) {
				count++;
			}
		}

		if (count != 1) {
			return "Only one action can be active. (" + count + ")";
		}
		return null;
	}

	public String getNameNode() {
		return nameNode;
	}

	public void setNameNode(String nameNode) {
		this.nameNode = nameNode;
	}

	public String getPath() {
		return path;
	}

	public void setPath(String path) {
		this.path = path;
	}

	public String getContent() {
		return content;
	}

	public void setContent(String content) {
		this.content = content;
	}

	public String getUpload() {
		return upload;
	}

	public void setUpload(String upload) {
		this.upload = upload;
	}

	public String getNewName() {
		return newName;
	}

	public void setNewName(String newName) {
		this.newName = newName;
	}

	public long getOffset() {
		return offset;
	}

	public void setOffset(long offset) {
		this.offset = offset;
	}

	public int getSize() {
		return size;
	}

	public void setSize(int size) {
		this.size = size;
	}

	public Boolean getCreateActive() {
		return createActive;
	}

	public void setCreateActive(Boolean createActive) {
		this.createActive = createActive;
	}

	public Boolean getRenameActive() {
		return renameActive;
	}

	public void setRenameActive(Boolean renameActive) {
		this.renameActive = renameActive;
	}

	public Boolean getDeleteActive() {
		return deleteActive;
	}

	public void setDeleteActive(Boolean deleteActive) {
		this.deleteActive = deleteActive;
	}

	public Boolean getTailActive() {
		return tailActive;
	}

	public void setTailActive(Boolean tailActive) {
		this.tailActive = tailActive;
	}

	public Boolean getOverwrite() {
		return overwrite;
	}

	public void setOverwrite(Boolean overwrite) {
		this.overwrite = overwrite;
	}

	public String getCreateCodec() {
		return createCodec;
	}

	public void setCreateCodec(String createCodec) {
		this.createCodec = createCodec;
	}

	public Boolean getRecursive() {
		return recursive;
	}

	public void setRecursive(Boolean recursive) {
		this.recursive = recursive;
	}

	public String getTailCodec() {
		return tailCodec;
	}

	public void setTailCodec(String tailCodec) {
		this.tailCodec = tailCodec;
	}

	public Boolean getInfo() {
		return info;
	}

	public void setInfo(Boolean info) {
		this.info = info;
	}

}
